package com.itheima.kafka.listener;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;

/*kafka消息监听者*/
@Component
public class HellpListencer {

    @KafkaListener(topics = {"kafka-hello"})
    public void getMsg(ConsumerRecord<?,?> consumerRecord){
        Optional<? extends ConsumerRecord<?, ?>> record = Optional.ofNullable(consumerRecord);

        //判断 不为空
        if (record.isPresent()){
            System.out.println(consumerRecord.value()+"1");
        }

    }

    @KafkaListener(topics = {"kafka-hi"})
    public  void getmsg(String msg){
        //接收 参数 并且 转换 为 json 对象
        System.out.println(msg);
        Map map = JSON.parseObject(msg, Map.class);
        System.out.println("map = " + map);
    }
}
